package com.atguigu.client;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.ColumnValueFilter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;

public class HbaseDML {
    public static Connection connection = Hbase_Single_Client.connection;

    public static void putCell(String nameSpace, String tableName, String rowkey, String family, String column, String value) throws IOException {
        //1.获取Table对象 //指定namespace 和 表名
        Table table = connection.getTable(TableName.valueOf(nameSpace, tableName));
        //2.创建put对象
        Put put = new Put(Bytes.toBytes(rowkey));
        //3.指定列族 列名 要写进去的数据
        put.addColumn(Bytes.toBytes(family), Bytes.toBytes(column), Bytes.toBytes(value));
        //4.执行put操作
        table.put(put);
        //关闭链接
        table.close();
    }

    public static void getCell(String nameSpace, String tableName, String rowkey, String family, String column) throws IOException {
        //1.获取Table对象
        Table table = connection.getTable(TableName.valueOf(nameSpace, tableName));

        //创建Get对象
        Get get = new Get(Bytes.toBytes(rowkey));

        //可以额外指定获取哪一列的数据
        get.addColumn(Bytes.toBytes(family), Bytes.toBytes(column));

        //执行get操作获取指定行或者指定列的数据
        Result result = table.get(get);
//        byte[] value = result.getValue(Bytes.toBytes(family), Bytes.toBytes(column));

//        System.out.println(Bytes.toString(value));

//        System.out.println(Bytes.toString(result.getRow()));
        for (Cell cell : result.rawCells()) {
            System.out.println("rowkey:"+ Bytes.toString(CellUtil.cloneRow(cell))+
                        " columnFamily:"+Bytes.toString(CellUtil.cloneFamily(cell))+
                               " column:"+Bytes.toString(CellUtil.cloneQualifier(cell))+
                               " value:"+Bytes.toString(CellUtil.cloneValue(cell))
                       );
        }

        //最后关闭链接
        table.close();
    }


    public static void deleteCell(String nameSpace, String tableName, String rowkey, String family, String column) throws IOException {
        //1.获取Table对象
        Table table = connection.getTable(TableName.valueOf(nameSpace, tableName));

        //创建delete对象
        Delete delete = new Delete(Bytes.toBytes(rowkey));

        delete.addColumn(Bytes.toBytes(family), Bytes.toBytes(column));

        //调用delete方法
        table.delete(delete);

        table.close();
    }

    public static void scanCells(String nameSpace, String tableName, String startRowkey, String stopRowKey) throws IOException {
        //1.获取Table对象
        Table table = connection.getTable(TableName.valueOf(nameSpace, tableName));

        //创建Scan
        Scan scan = new Scan();

        //获取所有版本的数据 包括超过版本维护的数据
        scan.setRaw(true);
        scan.readAllVersions();


        //指定扫描开始的rowkey  第二个参数代表的是是否包含指定行 开始行默认包含
        scan.withStartRow(Bytes.toBytes(startRowkey),true);
        //指定扫描结束的rowkey 第二个参数代表的是是否包含指定行 结束行默认不包含  可以改true则为包含
        scan.withStopRow(Bytes.toBytes(stopRowKey),true);

        //获取Scan
        ResultScanner scanner = table.getScanner(scan);
        for (Result result : scanner) {
            for (Cell cell : result.rawCells()) {
                System.out.println(
                       "rowkey:"+ Bytes.toString(CellUtil.cloneRow(cell))+
                        " columnFamily:"+Bytes.toString(CellUtil.cloneFamily(cell))+
                               " column:"+Bytes.toString(CellUtil.cloneQualifier(cell))+
                               " value:"+Bytes.toString(CellUtil.cloneValue(cell))
                        );
            }
        }
        table.close();
    }

    public static void scanFilterCells(String nameSpace, String tableName, String startRowkey, String stopRowKey,String family,String column,String value) throws IOException {
        //1.获取Table对象
        Table table = connection.getTable(TableName.valueOf(nameSpace, tableName));
        //创建Scan
        Scan scan = new Scan();

        //根据某个列具体的值过滤数据  过滤出 info:name 为zs的数据
        ColumnValueFilter columnValueFilter = new ColumnValueFilter(
                Bytes.toBytes(family),
                Bytes.toBytes(column),
                CompareOperator.EQUAL,
                Bytes.toBytes(value)
        );


        //创建一个Fileter过滤器的集合用来存放多个过滤器
        FilterList filterList = new FilterList();
        filterList.addFilter(columnValueFilter);


        //TODO 添加过滤器
        scan.setFilter(filterList);

        //获取Scan
        ResultScanner scanner = table.getScanner(scan);
        for (Result result : scanner) {
            for (Cell cell : result.rawCells()) {
                System.out.println(
                        "rowkey:"+ Bytes.toString(CellUtil.cloneRow(cell))+
                                " columnFamily:"+Bytes.toString(CellUtil.cloneFamily(cell))+
                                " column:"+Bytes.toString(CellUtil.cloneQualifier(cell))+
                                " value:"+Bytes.toString(CellUtil.cloneValue(cell))
                );
            }
        }
        table.close();
    }

    public static void main(String[] args) throws IOException {
//        String rowKey = null;
//        rowKey = user + Date;

//        putCell("bigdata","student","1001","info","age","18");
//        getCell("bigdata","student","1001","info","name");
//        deleteCell("bigdata","student","1001","info","age");
//        scanCells("bigdata","student","1001","1003");
        scanFilterCells("bigdata","student","1001","9999","info","name","zs");
    }
}
